package com.lcy.common;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunctions extends RichSinkFunction<JSONObject> {

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void close() throws Exception {
        conn.close();

    }

    @Override
    public void invoke(JSONObject jsonObject, Context context){
        PreparedStatement ps = null;
        try {
        JSONObject after = jsonObject.getJSONObject("after");
        String sinkTable = jsonObject.getString("sinkTable");
        String sinkPk = jsonObject.getString("sinkPk");
        String type = jsonObject.getString("type");

//        if ("update".equals(type)) {
//            //如果是修改的状态先删除redis里面的数据
//            Jedis jedis = RedisUtil.getJedis();
//            String key = sinkTable.toLowerCase() + ":" + after.getString(sinkPk);
//            jedis.del(key);
//            jedis.close();
//        }
          String upsertSQL = createUpsertSQL(sinkTable, after);
            System.out.println("=============" + upsertSQL);
            ps = conn.prepareStatement(upsertSQL);
            ps.executeUpdate();
            conn.commit();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            if (ps!=null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
        }


    }

    private String createUpsertSQL(String sinkTable, JSONObject after) {
        Set<String> strings = after.keySet();
        Collection<Object> values = after.values();
        String str = "upsert into "+GmallConfig.HBASE_SCHEMA + "."
                + sinkTable + "( " + StringUtils.join(strings, ",") + ")  values('" + StringUtils.join(values, "','") + "')";
        return str;
    }
}
